defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Ingester do
  @moduledoc """
  Simplified ingestion-only functionality for ClickHouse.
  """

  import Bitwise
  import Logflare.Utils.Guards

  alias Logflare.Backends.Backend
  alias Logflare.LogEvent

  @finch_pool Logflare.FinchClickHouseIngest
  @max_retries 3
  @initial_delay 500
  @max_delay 4_000

  @doc """
  Inserts a list of `LogEvent` structs into ClickHouse.

  Not intended for direct use. Use `Logflare.Backends.Adaptor.ClickHouseAdaptor.insert_log_events/2` instead.
  """
  @spec insert(Backend.t() | Keyword.t(), table :: String.t(), log_events :: [LogEvent.t()]) ::
          :ok | {:error, String.t()}
  def insert(_backend_or_conn_opts, _table, []), do: :ok

  def insert(%Backend{} = backend, table, log_events) when is_list(log_events) do
    with {:ok, connection_opts} <- build_connection_opts(backend) do
      insert(connection_opts, table, log_events)
    end
  end

  def insert(connection_opts, table, [%LogEvent{} | _] = log_events)
      when is_list(connection_opts) and is_non_empty_binary(table) do
    client = build_client(connection_opts)
    url = build_request_url(connection_opts, table)
    request_body = log_events |> encode_batch() |> :zlib.gzip()

    case Tesla.post(client, url, request_body) do
      {:ok, %Tesla.Env{status: 200}} ->
        :ok

      {:ok, %Tesla.Env{status: status, body: response_body}} ->
        {:error, "HTTP #{status}: #{response_body}"}

      {:error, reason} ->
        {:error, reason}
    end
  end

  @spec build_client(Keyword.t()) :: Tesla.Client.t()
  defp build_client(connection_opts) do
    middleware = [
      {Tesla.Middleware.Headers,
       [{"content-type", "application/octet-stream"}, {"content-encoding", "gzip"}]},
      {Tesla.Middleware.BasicAuth,
       %{
         username: Keyword.get(connection_opts, :username),
         password: Keyword.get(connection_opts, :password)
       }},
      {Tesla.Middleware.Retry,
       delay: @initial_delay,
       max_retries: @max_retries,
       max_delay: @max_delay,
       should_retry: &retriable?/1}
    ]

    adapter =
      {Tesla.Adapter.Finch, name: @finch_pool, pool_timeout: 4_000, receive_timeout: 8_000}

    Tesla.client(middleware, adapter)
  end

  @spec retriable?({:ok, Tesla.Env.t()} | {:error, term()}) :: boolean()
  defp retriable?({:ok, %Tesla.Env{status: status}}) when status >= 500, do: true
  defp retriable?({:ok, %Tesla.Env{status: 429}}), do: true
  defp retriable?({:ok, _env}), do: false
  defp retriable?({:error, _reason}), do: true

  @doc false
  @spec encode_row(LogEvent.t()) :: iodata()
  def encode_row(%LogEvent{body: body}) do
    [
      encode_as_uuid(body["id"]),
      encode_as_string(Jason.encode_to_iodata!(body)),
      encode_as_datetime64(DateTime.from_unix!(body["timestamp"], :microsecond))
    ]
  end

  @doc false
  @spec encode_batch([LogEvent.t()]) :: iodata()
  def encode_batch([%LogEvent{} | _] = rows) do
    Enum.map(rows, &encode_row/1)
  end

  @doc false
  @spec encode_as_uuid(Ecto.UUID.t() | String.t()) :: binary()
  def encode_as_uuid(uuid_string) when is_non_empty_binary(uuid_string) do
    uuid_raw =
      uuid_string
      |> String.replace("-", "")
      |> Base.decode16!(case: :mixed)

    case uuid_raw do
      <<u1::64, u2::64>> ->
        <<u1::64-little, u2::64-little>>

      _other ->
        raise "invalid uuid when trying to encode for ClickHouse: #{inspect(uuid_string)}"
    end
  end

  @doc false
  @spec encode_as_string(iodata()) :: iodata()
  def encode_as_string(value) when is_list(value) do
    length = IO.iodata_length(value)
    [encode_as_varint(length), value]
  end

  @doc false
  @spec encode_as_datetime64(DateTime.t()) :: binary()
  def encode_as_datetime64(%DateTime{microsecond: {microsecond, _precision}} = value) do
    timestamp_seconds = DateTime.to_unix(value, :second)
    timestamp_scaled = timestamp_seconds * 1_000_000 + microsecond
    <<timestamp_scaled::little-signed-64>>
  end

  @doc false
  @spec encode_as_varint(non_neg_integer()) :: binary()
  def encode_as_varint(n) when is_non_negative_integer(n) and n < 128, do: <<n>>

  def encode_as_varint(n) when is_non_negative_integer(n),
    do: <<1::1, n::7, encode_as_varint(n >>> 7)::binary>>

  @spec build_connection_opts(Backend.t()) :: {:ok, Keyword.t()} | {:error, String.t()}
  defp build_connection_opts(%Backend{
         config: %{
           url: url,
           port: port,
           database: database,
           username: username,
           password: password
         }
       }) do
    {:ok,
     [
       url: url,
       port: port,
       database: database,
       username: username,
       password: password
     ]}
  end

  defp build_connection_opts(_backend) do
    {:error, "Unable to build connection options"}
  end

  @spec build_request_url(connection_opts :: Keyword.t(), table :: String.t()) :: String.t()
  defp build_request_url(connection_opts, table) do
    base_url = Keyword.get(connection_opts, :url)
    database = Keyword.get(connection_opts, :database)

    uri = URI.parse(base_url)
    scheme = uri.scheme || "http"
    host = uri.host
    port = Keyword.get(connection_opts, :port, default_port(scheme))

    query = "INSERT INTO #{database}.#{table} FORMAT RowBinary"

    params =
      URI.encode_query(%{
        "query" => query,
        "async_insert" => "1",
        "wait_for_async_insert" => "1"
      })

    "#{scheme}://#{host}:#{port}/?#{params}"
  end

  defp default_port("https"), do: 8443
  defp default_port(_), do: 8123
end
